package com.shujia.streaming

import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo5StructuredStreaming {
  def main(args: Array[String]): Unit = {
    // 创建Spark的入口
    val spark: SparkSession = SparkSession
      .builder()
      .master("local[2]")
      .appName("Demo5StructedStreaming")
      .config("spark.sql.shuffle.partitions", 2)
      .getOrCreate()


    val lines: DataFrame = spark
      // 从流中读取数据
      .readStream
      .format("socket")
      .option("host", "master")
      .option("port", 8888)
      .load()

    // 打印表结构
    lines.printSchema()

    import spark.implicits._
    import org.apache.spark.sql.functions._


    /**
     * OutputMode:三种输出模式
     * Complete ：完整的输出，只能用于分组聚合之后的数据
     * Update ：只输出更新的数据
     * Append ： 追加，不能运用在分组聚合之后的数据
     */

    // DSL wordCount on streaming
    val wordDF: DataFrame = lines
      .select(explode(split($"value", ",")) as "word")

    wordDF.createOrReplaceTempView("words")

    // 批流合一
    //    spark.sql(
    //      """
    //        |select word,count(word) as c from words group by word
    //        |""".stripMargin)
    //      .writeStream
    //      .outputMode(OutputMode.Update()) // 只会输出更新的数据
    //      .format("console")
    //      .start()
    //      .awaitTermination()


    //    wordDF
    //      .groupBy($"word")
    //      .agg(count($"word") as "cnt") // 在StructedStreaming自带状态
    //      .writeStream
    //      .outputMode(OutputMode.Complete())
    //      .format("console")
    //      .start() // 启动流处理任务
    //      .awaitTermination() // 等待关闭

    // 将结果保存到MySQL
    spark.sql(
      """
        |select word,count(word) as cnt from words group by word
        |""".stripMargin)
      .writeStream
      .outputMode(OutputMode.Complete())
      // 将流转换成DataSet进行处理
      .foreachBatch((df, l) => {
        // 表结构会发生变化
        // 字符类型的数据 到了MySQL默认会被Spark修改为text类型
        // 整型的数据 到了MySQL默认会被Spark修改为bigint类型
        df.write
          .mode(SaveMode.Overwrite)
          .format("jdbc")
          .option("url", "jdbc:mysql://master:3306/student")
          .option("dbtable", "word_cnt")
          .option("user", "root")
          .option("password", "123456")
          .save()
      })
      .start()
      .awaitTermination()

  }

}
